package major;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

public class CountInter implements ProducerInterceptor<String, String> {
    private AtomicInteger count = new AtomicInteger(0);

    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> producerRecord) {
        debug(producerRecord);
//        main : 0_count  org.apache.kafka.clients.producer.ProducerRecordProducerRecord
        return producerRecord;
    }

    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
        count.getAndIncrement();
        debug(recordMetadata);
        //    kafka-producer-network-thread | producer-1 : 53_count
//    org.apache.kafka.clients.producer.RecordMetadata
    }

    @Override
    public void close() {
        debug("结束了");
//        Thread-0 : 53_count  java.lang.String结束了
    }

    @Override
    public void configure(Map<String, ?> map) {

    }

    public void debug(Object o) {

        System.out.println(Thread.currentThread().getName() + " : " +
                count + "_count  " + o.getClass().getName() + o
        );
    }
}